package org.apache.ocean.solr.core;

import java.io.IOException;
import java.util.HashMap;
import java.util.HashSet;
import java.util.List;
import java.util.Map;
import java.util.NavigableMap;
import java.util.Set;
import java.util.TreeMap;

import org.apache.commons.lang.ArrayUtils;
import org.apache.lucene.analysis.Analyzer;
import org.apache.lucene.document.Document;
import org.apache.lucene.index.FilterIndexReader;
import org.apache.lucene.index.IndexReader;
import org.apache.lucene.store.instantiated.InstantiatedIndex;
import org.apache.lucene.store.instantiated.InstantiatedIndexReader;
import org.apache.lucene.store.instantiated.InstantiatedIndexWriter;
import org.apache.ocean.main.IndexDocuments.IndexDocument;
import org.apache.ocean.main.search.cache.Cache;
import org.apache.ocean.main.search.cache.CacheKey;
import org.apache.ocean.main.search.cache.DocSetCacheKey;
import org.apache.ocean.main.search.cache.FieldCacheKey;
import org.apache.ocean.main.search.cache.FieldCacheKey.FieldIndex;
import org.apache.ocean.main.search.cache.FieldCacheKey.FieldIndex.ValueIndex;
import org.apache.ocean.main.search.cache.FieldCacheKey.FieldIndex.ValueIndexComparator;
import org.apache.ocean.main.util.Constants;
import org.apache.ocean.main.util.Timeout;
import org.apache.ocean.util.BitDocSet;
import org.apache.ocean.util.ConcurrentTreeMap;

// TODO: release old MemoryIndexSnapshots from map
public class WriteableMemoryIndex extends Index {
	private WriteableIndexWriter indexWriter;
	private InstantiatedIndexReader indexReader;
	private InstantiatedIndex instantiatedIndex;
	private ConcurrentTreeMap<Long, MemoryIndexSnapshot> snapshotMap = new ConcurrentTreeMap<Long, MemoryIndexSnapshot>();
	private Cache cache;

	public WriteableMemoryIndex(Long id, Analyzer defaultAnalyzer, TransactionSystem.Category category) throws IOException {
		super(id, defaultAnalyzer, category);
		instantiatedIndex = new InstantiatedIndex();
		indexWriter = new WriteableIndexWriter(instantiatedIndex);
		indexReader = new InstantiatedIndexReader(instantiatedIndex);
		cache = initCache();
	}

	// called by Category.runTransactionsNotInIndex
	void addDocuments(IndexDocuments indexDocuments, Analyzer analyzer) throws IOException {
		for (IndexDocument indexDocument : indexDocuments) {
			Document document = indexDocument.toDocument();
			indexWriter.addDocument(document, analyzer);
		}
	}

	// called by Category.runTransactionsNotInIndex
	MemoryIndexSnapshot setSnapshot(Long snapshotID, List<Deletes> deletesList, IndexSchema.Category schemaCategory) throws Exception {
		int maxDoc = indexReader.maxDoc();
		HashSet<Integer> deletedSet = new HashSet<Integer>();
		if (deletesList != null) {
			for (Deletes deletes : deletesList) {
				FieldIndex idFieldIndex = cache.getFieldCacheData(new FieldCacheKey(Constants.ID, Long.class)).getObject();
				applyDeletes(false, deletes, null, indexReader, idFieldIndex);
			}
		}
		MemoryIndexSnapshot memoryIndexSnapshot = new MemoryIndexSnapshot(snapshotID, maxDoc, deletedSet, schemaCategory);
		snapshotMap.put(snapshotID, memoryIndexSnapshot);
		return memoryIndexSnapshot;
	}

	public MemoryIndexSnapshot getIndexSnapshot(Long snapshotID) {
		return snapshotMap.get(snapshotID);
	}

	public MemoryIndexSnapshot getLatestIndexSnapshot() {
		return snapshotMap.lastValue();
	}

	public class MemoryIndexSnapshot extends IndexSnapshot {
		private final int maxDoc;
		private HashSet<Integer> deletedDocs;
		private SnapshotIndexReader snapshotIndexReader;

		public MemoryIndexSnapshot(Long snapshotID, int maxDoc, HashSet<Integer> deletedDocs, IndexSchema.Category schemaCategory) {
			super(snapshotID, schemaCategory);
			this.maxDoc = maxDoc;
			this.deletedDocs = deletedDocs;
		}

		public FieldIndex createFieldIndex(String field, Timeout timeout) throws Exception {
			return cache.getFieldIndex(field, this, schemaCategory);
		}
		
		public FieldIndex getFieldIndex(String field) throws Exception {
			return cache.getFieldIndex(field, this, schemaCategory);
		}
   
		private Long getHighest(String field) throws Exception {
			FieldIndex fieldIndex = getFieldIndex(field);
			if (fieldIndex == null) return null;
			ValueIndex valueIndex = fieldIndex.valueIndices[fieldIndex.valueIndices.length-1];
			return (Long)valueIndex.value;
		}
		
		public boolean containsKey(CacheKey cacheKey) {
			return cache.containsKey(cacheKey);
		}
		
		public Long getHighestSnapshotId() throws Exception {
			return getHighest(Constants.SNAPSHOTID);
		}

		public Long getHighestDocumentId() throws Exception {
			return getHighest(Constants.ID);
		}

		public Set<CacheKey> getCacheKeys() {
			return cache.getCacheKeys();
		}

		public IndexReader getIndexReader() {
			return snapshotIndexReader;
		}

		public class SnapshotIndexReader extends FilterIndexReader {
			private int numDocs;
			
			public SnapshotIndexReader(int numDocs) {
				super(indexReader);
				this.numDocs = numDocs;
			}

			public boolean isDeleted(int doc) {
				if (doc >= maxDoc) return true;
				return deletedDocs.contains(doc);
			}

			public boolean hasDeletions() {
				return true;
			}

			public int numDocs() {
				return numDocs;
			}

			public int maxDoc() {
				return maxDoc;
			}
		}

		public BitDocSet getDocSet(DocSetCacheKey cacheKey, Timeout timeout) throws Exception {
			return cacheKey.createObject(this, timeout);
		}
	}

	Long getLatestSnapshotID() {
		return snapshotMap.lastKey();
	}

	private HashSet<Integer> getLatestSnapshotDeletedDocSet() {
		MemoryIndexSnapshot memoryIndexSnapshot = snapshotMap.lastValue();
		if (memoryIndexSnapshot == null || memoryIndexSnapshot.deletedDocs == null)
			return null;
		HashSet<Integer> deletedDocSet = memoryIndexSnapshot.deletedDocs;
		return deletedDocSet;
	}

	public int getDocumentCount() {
		return indexReader.maxDoc();
	}

	public DeletesResult commitDeletes(Deletes deletes, Transaction transaction) throws InterruptedException, IOException {
		return commitChanges(null, deletes, null, transaction);
	}

	public DeletesResult commitChanges(IndexDocuments indexDocuments, Deletes deletes, Analyzer analyzer, Transaction transaction) throws InterruptedException, IOException {
		try {
			if (isClosed()) {
				throw new IOException("index is closed");
			}
			if (isReadOnly() && indexDocuments != null && indexDocuments.size() > 0) {
				throw new IOException("index not accepting new documents");
			}
			DeletesResult deletesResult = new DeletesResult(getId());
			HashSet<Integer> deletedSet = null;
			if (deletes != null && deletes.hasDeletes()) {
				HashSet<Integer> previousDeletedSet = getLatestSnapshotDeletedDocSet();
				if (previousDeletedSet != null) {
					deletedSet = (HashSet<Integer>) previousDeletedSet.clone();
				} else {
					deletedSet = new HashSet<Integer>();
				}
				IndexSnapshot indexSnapshot = this.getLatestIndexSnapshot();
				FieldIndex idFieldIndex = indexSnapshot.getFieldIndex(Constants.ID);
				
				deletesResult = applyDeletes(false, deletes, deletedSet, indexReader, idFieldIndex);
			} else if (deletes == null || !deletes.hasDeletes()) { // if no deletes
				// just use same
				deletedSet = getLatestSnapshotDeletedDocSet();
			}
			if (indexDocuments != null) {
				for (IndexDocument document : indexDocuments) {
					indexWriter.addDocument(document.toDocument(), analyzer);
				}
			}
			transaction.ready(this);
			if (transaction.go()) {
				indexWriter.commit();
				int maxDoc = indexReader.maxDoc();
				Long snapshotId = transaction.getId();
				// recreate field cache
				if (indexDocuments != null) {
					int doc = maxDoc - indexDocuments.size() - 1;
					IndexSchema.Category schemaCategory = transaction.getIndexSchemaCategory();
					List<String> indexableFieldNames = schemaCategory.getFieldIndexableNames();
					Map<String,NavigableMap<Object, ValueIndex>> newValueIndices = new HashMap<String,NavigableMap<Object, ValueIndex>>(indexableFieldNames.size());
					for (IndexDocument document : indexDocuments) {
						IndexSnapshot previousIndexSnapshot = getLatestIndexSnapshot();
						for (String fieldName : indexableFieldNames) {
							Object value = document.get(fieldName);
							if (value != null) {
								FieldIndex fieldIndex = previousIndexSnapshot.getFieldIndex(fieldName);
								ValueIndex valueIndex = (ValueIndex) fieldIndex.getSortedMap().get(value);
								int[] valueIndexDocs = null;
								if (valueIndex == null) {
									valueIndexDocs = new int[] {doc}; 
								} else {
									valueIndexDocs = ArrayUtils.add(valueIndex.docs, doc);
								}
								NavigableMap<Object, ValueIndex> newSortedMap = newValueIndices.get(fieldName);
								if (newSortedMap == null) {
									newSortedMap = new TreeMap<Object, ValueIndex>(new ValueIndexComparator());
									newValueIndices.put(fieldName, newSortedMap);
								}
								ValueIndex newValueIndex = new ValueIndex((Comparable)value, valueIndexDocs);
								newSortedMap.put(value, newValueIndex);
							}
						}
						doc++;
					}
					for (Map.Entry<String,NavigableMap<Object, ValueIndex>> entry : newValueIndices.entrySet()) {
						String field = entry.getKey();
						Class type = schemaCategory.getField(field).getFieldType().getType();
						FieldIndex fieldIndex = new FieldIndex(field, type, maxDoc, entry.getValue());
						cache.replaceFieldIndex(fieldIndex);
					}
				}
				MemoryIndexSnapshot memoryIndexSnapshot = new MemoryIndexSnapshot(snapshotId, maxDoc, deletedSet, transaction.getIndexSchemaCategory());
				snapshotMap.put(transaction.getId(), memoryIndexSnapshot);
				return deletesResult;
			} else {
				indexWriter.abort();
				return null;
			}
		} catch (Throwable throwable) {
			transaction.failed(this, throwable);
		}
		return null;
	}

	public class WriteableIndexWriter extends InstantiatedIndexWriter {
		public WriteableIndexWriter(InstantiatedIndex index) throws IOException {
			super(index);
		}
	}
}
